package com.atguigu.chapter05.sink;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.http.HttpHost;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.ArrayList;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/6/10 10:09
 */
public class Flink03_Sink_Custom {
    public static void main(String[] args) throws Exception {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        
        ArrayList<HttpHost> hosts = new ArrayList<>();
        hosts.add(new HttpHost("hadoop162", 9200));
        hosts.add(new HttpHost("hadoop163", 9200));
        hosts.add(new HttpHost("hadoop164", 9200));
        
        env
            .socketTextStream("hadoop162", 9999)
            .map(new MapFunction<String, WaterSensor>() {
                @Override
                public WaterSensor map(String line) throws Exception {
                    String[] data = line.split(",");
                    return new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2]));
                }
            })
            .keyBy(WaterSensor::getId)
            .sum("vc")
            .addSink(new MysqlSink());
        
        env.execute();
        
    }
    
    public static class MysqlSink extends RichSinkFunction<WaterSensor> {
        
        private Connection conn;
        private PreparedStatement ps;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            // 建立到MySQL的连接
            Class.forName("com.mysql.jdbc.Driver");
            //            conn = DriverManager.getConnection("jdbc:mysql://hadoop162:3306/test", "root", "aaaaaa");
            conn = DriverManager.getConnection("jdbc:mysql://hadoop162:3306/test?user=root&password=aaaaaa");
            //            String sql = "insert into sensor(id, ts, vc) values(?,?,?)";
            //String sql = "insert into sensor(id, ts, vc) values(?,?,?) on duplicate key update vc=?";
            String sql = "replace into sensor(id, ts, vc) values(?,?,?)";
            // 输出使用幂等性 或者 支持事务
            ps = conn.prepareStatement(sql);
        }
        
        @Override
        public void invoke(WaterSensor value,
                           Context context) throws Exception {
            // 进入一条数据
            ps.setString(1, value.getId());
            ps.setLong(2, value.getTs());
            ps.setInt(3, value.getVc());
//            ps.setInt(4, value.getVc());
            
            ps.execute();  // execute(ddl dml) executeQuery(查询)
            
        }
        
        @Override
        public void close() throws Exception {
            if (ps != null) {
                ps.close();
                
            }
            
            if (conn != null) {
                conn.close();
                
            }
        }
    }
}
